#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <time.h>
#include <errno.h>
#include <pthread.h>
#include <sys/stat.h>
#include <sys/ipc.h>
#include <sys/msg.h>

#include "rsc.h"
#include "debug.h"
#include "record.h"
//#include "sdmng.h"
//#include "exception.h"

#define MAX_VBUFF_SIZE (400*1024)
#define MAX_ABUFF_SIZE 1024
#define MAX_UDATA_SIZE 1024

enum _rsc_msg_cmd
{
    CMD_START_CHN = 11,
    CMD_STOP_CHN,
    CMD_GETATTR_CHN,
    CMD_SETATTR_CHN,
    CMD_DESTROY_CHN,
    CMD_GETSTAT_CHN,

    CMD_RES_START = 101,
    CMD_RES_GETATTR,
    CMD_RES_SETATTR,
    CMD_RES_GETSTAT,
};

struct _rsc_chn_thread
{
    int 			chn;
    pthread_t			ctid;
    struct record_sink		csink;
    struct rsc_channel_attr	cattr;
    struct rsc_channel_stat	cstat;
};

struct _rsc_msg_item
{
    long msgtype;
    int mpsize;
    char payload[1024];
};

/******************* Data Variables used in RSC-record module *************/
#if  0 
static void print_helper(const char *str,unsigned char *buf,int len)
{
    printf("\n%s:%d\n",str, len);
    int i = 0;
    for(i=1; i<=len; i++)
    {
	printf("0x%02x ",buf[i-1]);
	if(i%16 == 0)
	    printf("\n");
    }
    printf("\n");
}
#endif

static int run_rec = 0; 
static int rsc_rec_flag = 0; /* max 32 chns, from 0~31, 1 - recon, 0 - recoff */
static int rsc_chn_flag = 0; /* max 32 chns, from 0~31, 1 - use, 0 - unuse*/
static int rsc_env_stat = 0;
static pthread_mutex_t rsc_rec_lock = PTHREAD_MUTEX_INITIALIZER;

static int count_recsink_size(int brate, int time)
{
    int size = 0;

    size = ((unsigned long long)brate * time/8)>>20; // MB

    if(size == 0)
	size = 1;

    return size;
}

static int init_record_sink(struct record_sink *recsink,struct rsc_channel_attr *recattr)
{
    long brate = recattr->vstream.brate + recattr->astream.brate + recattr->mstream.brate;
    int size = count_recsink_size(brate,recattr->unit_time_limit);

    /** init all members */
    struct rsc_stream_confset media;
    memset(recsink->mpoint,0,sizeof(recsink->mpoint));
    memset(recsink->sname,0,sizeof(recsink->sname));
    memset(recsink->user_data,0,sizeof(recsink->user_data));
    memset(recsink->audio_buf,0,sizeof(recsink->audio_buf));
    recsink->ctime = time(NULL);
    recsink->ssize = 0;
    recsink->psink = NULL;
    recsink->user_len = 0;
    recsink->video_buf = NULL;

    SLOG(PRI,"estimate file size:[%d]MB,total:x%d = %d MB,tlen = %dSecs\n", 
	    size,2,size*2,recattr->unit_time_limit);
#if 1
    strcpy(recsink->mpoint,"/home/lzhong/test/rscsd"); // debug
#else
    /** we suppose all channels is recording to ensure space is ok */
    if(rsc_get_avail_dev(size*2, recsink->mpoint)<0)
    {
	SLOG(ERR,"rsc_get_avail_dev error!\n");
	rsc_rec_setpenv(0);
	return -1;
    }
#endif

    memset(&media,0,sizeof(media));
    if(recattr->vstream.on)
    {
	media.strset = RSC_STREAM_VIDEO;
	media.v_codec = recattr->vstream.codec;
	media.v_cbr = recattr->vstream.cbr;
	media.v_brate = recattr->vstream.brate;
	media.v_frate = recattr->vstream.frate;
	media.v_gop = recattr->vstream.gop;
	media.v_width = recattr->vstream.width;
	media.v_height = recattr->vstream.height;
	media.v_pic_level = recattr->vstream.pic_level;
    }
    if(recattr->astream.on)
    {
	media.strset |= RSC_STREAM_AUDIO;
	media.a_codec = recattr->astream.codec;
	media.a_sample_fmt = recattr->astream.sample_fmt;
	media.a_sample_rate = recattr->astream.sample_rate;
	media.a_brate = recattr->astream.brate;
	media.a_channels = recattr->astream.channels;
    }
    if(recattr->mstream.on)
    {
	media.strset |= RSC_STREAM_META;
	media.m_brate = recattr->mstream.brate;
    }

    recsink->psink = recattr->sink.init_sink(recattr->chn,recsink,&media,&recattr->modeattr);
    if(recsink->psink == NULL)
    {
	SLOG(ERR,"init_sink fail!\n");
	return -1;
    }
    recsink->video_buf = (unsigned char *)malloc(MAX_VBUFF_SIZE*sizeof(char));
    if(recsink->video_buf==NULL)
    {
	SLOG(ERR,"malloc recsink videobuf[%d Bytes] fail!\n",MAX_VBUFF_SIZE);
	return -1;
    }
    SLOG(PRI,"recsink name: %s\n",recsink->sname);
    return 0;
}

static int save_record_param(struct rsc_channel_attr *param)
{
    int ret = 0;
    FILE *pf = NULL;

    if(param == NULL)
    {
	SLOG(ERR, "invalid parameter\n");
	return -1;
    }
    pthread_mutex_lock(&rsc_rec_lock);
    pf = fopen(RECORD_CONFIG_FILE, "w");
    if (pf == NULL) 
    {
	SLOG(ERR, "fopen %s err\n", RECORD_CONFIG_FILE);
	pthread_mutex_unlock(&rsc_rec_lock);
	return -1;
    }
    ret = fwrite(param, sizeof(*param), 1, pf);
    if (ret != 1) 
    {
	SLOG(ERR, "fwrite to %s err\n", RECORD_CONFIG_FILE);
	fclose(pf);
	pthread_mutex_unlock(&rsc_rec_lock);
	return -1;
    }
    fclose(pf);
    pthread_mutex_unlock(&rsc_rec_lock);

    return 0;
}

static int read_record_param(struct rsc_channel_attr *param)
{
    int ret = 0;
    FILE *pf = NULL;

    if(param == NULL)
    {
	SLOG(ERR, "invalid parameter\n");
	return -1;
    }

    pthread_mutex_lock(&rsc_rec_lock);
    pf = fopen(RECORD_CONFIG_FILE, "r");
    if (pf == NULL) 
    {
	SLOG(ERR, "fopen %s err\n", RECORD_CONFIG_FILE);
	pthread_mutex_unlock(&rsc_rec_lock);
	return -1;
    }

    ret = fread(param, sizeof(*param), 1, pf);
    if (ret != 1) 
    {
	SLOG(ERR, "fread to %s err\n", RECORD_CONFIG_FILE);
	fclose(pf);
	pthread_mutex_unlock(&rsc_rec_lock);
	return -1;
    }

    fclose(pf);
    pthread_mutex_unlock(&rsc_rec_lock);
    return 0;
}

int rsc_rec_getconf(struct rsc_channel_attr *conf)
{
    if(conf==NULL)
	return -1;
    return read_record_param(conf);
}

int rsc_rec_setconf(struct rsc_channel_attr *conf,int mask)
{
    int ret = 0,i;

    if(conf==NULL)
	return -1;

    ret = save_record_param(conf);
    for(i=0;i<sizeof(rsc_chn_flag)*8;i++)
    {
	if((rsc_chn_flag&mask)&(1<<i))
	{
	    rsc_chn_setattr(i,conf);
	}
    }
    return ret;
}

void rsc_rec_setflag(int chn,int flag)
{
    pthread_mutex_lock(&rsc_rec_lock);
    if(((rsc_rec_flag>>chn)&1) != flag)
    {
	rsc_rec_flag &= ~(1<<chn);
	rsc_rec_flag |= (flag<<chn);
    }
    pthread_mutex_unlock(&rsc_rec_lock);
}

int rsc_rec_isrecod(void)
{
    int on = 0;

    pthread_mutex_lock(&rsc_rec_lock);
    on = rsc_env_stat && rsc_rec_flag;
    pthread_mutex_unlock(&rsc_rec_lock);
    return on;
}

int rsc_rec_getpenv(void)
{
    int on = 0;

    pthread_mutex_lock(&rsc_rec_lock);
    on = rsc_env_stat; 
    pthread_mutex_unlock(&rsc_rec_lock);
    return on;
}
int rsc_rec_setpenv(int flag)
{
    int on = 0;

    pthread_mutex_lock(&rsc_rec_lock);
    on = rsc_env_stat;
    rsc_env_stat = flag; 
    pthread_mutex_unlock(&rsc_rec_lock);
    return on;
}
int rsc_rec_start(int mask)
{
    int i = 0;
    for(i=0;i<sizeof(rsc_chn_flag)*8;i++)
    {
	if((rsc_chn_flag&mask)&(1<<i))
	{
	    rsc_chn_start(i);
	}
    }
    return 0;
}
int rsc_rec_stop(int mask)
{
    int i = 0;
    for(i=0;i<sizeof(rsc_chn_flag)*8;i++)
    {
	if((rsc_chn_flag&mask)&(1<<i))
	{
	    rsc_chn_stop(i);
	}
    }
    return 0;
}

int rsc_rec_init(void)
{
    rsc_rec_setpenv(1);
    run_rec = 1;
    return 0;
}
int rsc_rec_exit(void)
{
    int i = 0;
    for(i=0;i<sizeof(rsc_chn_flag)*8;i++)
    {
	if(rsc_chn_flag&(1<<i))
	{
	    rsc_chn_stop(i);
	    rsc_chn_destroy(i);
	}
    }
    while(rsc_chn_flag) 
	usleep(300000);
    run_rec = 0;
    return 0;
}

static int check_msg_queue(int chn)
{
    int msgid;
    key_t msgkey = ftok(".",chn);
    if((msgid=msgget(msgkey,0))<0)
	return -1;
    return msgid;
}
static int create_msg_queue(int chn)
{
    int msgid;

    key_t msgkey = ftok(".",chn);
    if((msgid=msgget(msgkey,0))<0)
    {
	msgid = msgget(msgkey,IPC_CREAT|0777);
	if(msgid < 0)
	    SLOG(ERR,"create chn[%d] msgqueue fail!\n",chn);
    }
    return msgid;
}
static int destroy_msg_queue(int chn)
{
    int msgid;

    key_t msgkey = ftok(".",chn);
    if((msgid=msgget(msgkey,0))<0)
    {
	return -1;
    }
    else
    {
	msgctl(msgid,IPC_RMID,0);
	return 0;
    }
}
static int send_msg_cmd(int chn,int cmd,void *buf,int size)
{
    int msgid,ret;
    if((msgid=check_msg_queue(chn))<0)
    {
	SLOG(ERR,"chn[%d] msgqueue doesnot exist,msgid = %d!\n",chn,msgid);
	return -1;
    } 
    struct _rsc_msg_item msg;
    if(size < 0 || size > sizeof(msg.payload))
    {
	SLOG(ERR,"msg payload size[%d] too big!\n",size);
	return -1;
    }
    msg.msgtype = cmd;
    msg.mpsize = size;
    if(size>0)
    {
	memcpy(msg.payload,buf,size);
    }

    ret = msgsnd(msgid,&msg,size+sizeof(int),IPC_NOWAIT);
    return ret;
}
static int recv_msg_cmd(int chn,int *cmd,void *buf,int *size)
{
    int msgid,ret;
    if((msgid=check_msg_queue(chn))<0)
    {
	SLOG(ERR,"chn[%d] msgqueue doesnot exist,msgid = %d!\n",chn,msgid);
	return -1;
    } 
    struct _rsc_msg_item msg;
    ret = msgrcv(msgid,&msg,sizeof(msg) - sizeof(long),*cmd,IPC_NOWAIT);
    if(ret<0)
    {
	//SLOG(ERR,"msgrcv chn[%d] fail!\n",chn);
	return -1;
    }
    *cmd = msg.msgtype;
    if(*size >= msg.mpsize)
    {
	memcpy(buf,msg.payload,msg.mpsize);
	*size = msg.mpsize;
	return 0;
    }
    else
    {
	SLOG(ERR,"msgsize[%d] > bufsize[%d]!\n",msg.mpsize,*size);
	return -1;
    }
}

static int process_msg_cmd(int chn,struct _rsc_chn_thread *prec)
{
    if(prec==NULL)
	return -1;

    int cmd,ret;
    char buf[1024];
    int size = sizeof(buf);

    cmd = CMD_RES_START*(-1);
    ret = recv_msg_cmd(chn,&cmd,buf,&size);
    if(ret < 0)
	return -1;
    switch(cmd)
    {
	case CMD_START_CHN:
	    printf(">>>recv CMD_START_CHN!\n");
	    prec->cattr.onstart = 1;
	    break;
	case CMD_STOP_CHN:
	    printf(">>>recv CMD_STOP_CHN!\n");
	    prec->cattr.onstart = 0;
	    break;
	case CMD_DESTROY_CHN:
	    printf(">>>recv CMD_DESTROY_CHN!\n");
	    prec->cattr.runflag = 0;
	    break;
	case CMD_GETATTR_CHN:
	    printf(">>>recv CMD_GETATTR_CHN!\n");
	    if(size >= sizeof(struct rsc_channel_attr))
	    {
		size = sizeof(struct rsc_channel_attr);
		cmd = CMD_RES_GETATTR;
		memcpy(buf,(void*)&prec->cattr,size);
		ret = send_msg_cmd(chn,cmd,buf,size);
	    }
	    else
	    {
		SLOG(ERR,"buf size[%d] < ressize[%d]!\n",size,sizeof(struct rsc_channel_attr));
		ret = -1;
	    }
	    break;
	case CMD_SETATTR_CHN: /* only dynamic attribute can be set */
	    {
		printf(">>>recv CMD_SETATTR_CHN!\n");
		struct rsc_channel_attr *pattr = (struct rsc_channel_attr *)buf;
		prec->cattr.vstream.on = pattr->vstream.on;
		prec->cattr.astream.on = pattr->astream.on;
		prec->cattr.mstream.on = pattr->mstream.on;
	    }
	    break;
	case CMD_GETSTAT_CHN:
	    printf(">>>recv CMD_GETSTAT_CHN!\n");
	    if(size >= sizeof(struct rsc_channel_stat))
	    {
		size = sizeof(struct rsc_channel_stat);
		cmd = CMD_RES_GETSTAT;
		memcpy(buf,(void*)&prec->cstat,size);
		ret = send_msg_cmd(chn,cmd,buf,size);
	    }
	    else
	    {
		SLOG(ERR,"buf size[%d] < ressize[%d]!\n",size,sizeof(struct rsc_channel_stat));
		ret = -1;
	    }
	    break;
	default:
	    SLOG(ERR,"unknown cmd[%d]\n",cmd);
	    break;
    }
    return ret;
}

static void *rsc_rec_thread(void *arg)
{
    time_t time_bak,time_now;
    int ret = 0,write_ing_flag = 0;
    //int wv_ret,wa_ret;
    int stream_on = 0;

    struct record_frame_info frame;
    struct _rsc_chn_thread *prec = (struct _rsc_chn_thread*)arg;

    SLOG(PRI,"enter record chn thread:%d\n", prec->chn);
    time_bak = time_now = time(NULL);
    prec->cstat.chn = prec->chn;
    while(run_rec && prec->cattr.runflag)
    {
	prec->cstat.runing = prec->cattr.runflag;
	prec->cstat.recing = write_ing_flag;
	process_msg_cmd(prec->chn,prec);

	stream_on = prec->cattr.vstream.on || prec->cattr.astream.on || prec->cattr.mstream.on;
	rsc_rec_setflag(prec->chn,stream_on);
	if(!prec->cattr.onstart || !rsc_rec_getpenv() || !stream_on)
	{
	    if(write_ing_flag) 
		goto sink_unit_end;
	    usleep(300000);
	    continue;
	}

	if(!write_ing_flag)
	{
	    if(init_record_sink(&prec->csink,&prec->cattr)<0)
	    {
		SLOG(ERR,"init record sink fail!\n");
		usleep(100000);
		continue;
	    }
	    else
	    {
		time_bak = prec->csink.ctime;
		write_ing_flag = 1;
		//wv_ret = wa_ret = 0;
		memcpy(prec->cstat.mpoint,prec->csink.mpoint,64);
		memcpy(prec->cstat.abspath,prec->csink.sname,128);

		//rsc_exception_handle(prec->chn,CONT_STA_INIT,prec->csink.mpoint,prec->csink.sname);
		//rsc_exception_handle(prec->chn,CONT_STA_WING,prec->csink.mpoint,prec->csink.sname);
	    }
	}

	time_now = time(NULL);
	if((time_now+10 < time_bak) || ((time_now - time_bak) >= prec->cattr.unit_time_limit) 
		|| (prec->csink.ssize >= prec->cattr.unit_size_limit))
	{
	    if(prec->cattr.chn_temp)
	    {
		prec->cattr.onstart  = 0;
	    }
	    ret = 0;
	    goto sink_unit_end;
	}
	else
	{
	    if(prec->cattr.vstream.on && prec->cattr.vstream.dsrc)
	    {
		memset(&frame,0,sizeof(frame));
		frame.addr = prec->csink.video_buf;
		frame.size = MAX_VBUFF_SIZE;
		prec->cattr.vstream.preprocess(&frame, prec->cattr.vstream.dsrc);
		if (frame.flag==0 || frame.len <= 0)
		{
		    ;/// at most case indicates source is empty
		}
		else if((ret=prec->cattr.sink.fill_sink(prec->csink.psink,RSC_STREAM_VIDEO,&frame))<0)
		{
		    SLOG(ERR, "write video frame err,ret = %d!\n",ret);
		    goto sink_unit_end;
		}
		else
		    prec->csink.ssize += frame.len;
	    }
	    if(prec->cattr.astream.on && prec->cattr.astream.dsrc)
	    {
		memset(&frame,0,sizeof(frame));
		frame.addr = prec->csink.audio_buf;
		frame.size = MAX_ABUFF_SIZE;
		prec->cattr.astream.preprocess(&frame, prec->cattr.astream.dsrc);
		if (frame.flag==0 || frame.len <= 0)
		{
		    ;/// at most case indicates source is empty
		}
		else if((ret=prec->cattr.sink.fill_sink(prec->csink.psink,RSC_STREAM_AUDIO,&frame))<0)
		{
		    SLOG(ERR, "write audio frame err,ret = %d!\n",ret);
		    goto sink_unit_end;
		}
		else
		    prec->csink.ssize += frame.len;
	    }
	    if(prec->cattr.mstream.on && prec->cattr.mstream.dsrc)
	    {
		memset(&frame,0,sizeof(frame));
		frame.addr = prec->csink.user_data;
		frame.size = MAX_UDATA_SIZE;
		prec->cattr.mstream.preprocess(&frame, prec->cattr.mstream.dsrc);
		if (frame.flag==0 || frame.len <= 0)
		{
		    ;/// at most case indicates source is empty
		}
		else if((ret=prec->cattr.sink.fill_sink(prec->csink.psink,RSC_STREAM_META,&frame))<0)
		{
		    SLOG(ERR, "write meta frame err,ret = %d!\n",ret);
		    goto sink_unit_end;
		}
		else
		    prec->csink.ssize += frame.len;
	    }
	}

	usleep(10000);
	continue;

sink_unit_end:
	write_ing_flag = 0;
	//if(ret == RSC_EWRIT)
	//    rsc_exception_handle(prec->chn,DEV_STA_WRITE_ERR,prec->csink.mpoint,prec->csink.sname);
	//rsc_exception_handle(prec->chn,CONT_STA_WEND,prec->csink.mpoint,prec->csink.sname);

	if(prec->csink.video_buf)
	{
	    free(prec->csink.video_buf);
	    prec->csink.video_buf = NULL;
	}
	prec->cattr.sink.exit_sink(prec->csink.psink);
	memset(prec->cstat.mpoint,0,64);
	memset(prec->cstat.abspath,0,128);
	usleep(300000);
    }
    
    SLOG(PRI,"exit record chn thread:%d\n", prec->chn);
    destroy_msg_queue(prec->chn);
    rsc_chn_flag &= ~(1<<prec->chn);
    if(prec!=NULL)
    {
	free(prec);
	prec = NULL;
    }
    return NULL;
}

int rsc_chn_create(int chn,struct rsc_channel_attr *attr)
{
    if(chn<0 || attr==NULL)
	return -1;

    int ret = 0;
    pthread_mutex_lock(&rsc_rec_lock);
    if(rsc_chn_flag&(1<<chn))
    {
	SLOG(ERR,"record chn[%d] has been created!\n",chn);
	ret = -1;
    }
    else if(create_msg_queue(chn)<0)
    {
	ret = -1;
    }
    else rsc_chn_flag |= (1<<chn);
    pthread_mutex_unlock(&rsc_rec_lock);
    if(ret < 0) return -1;

    struct _rsc_chn_thread *pchn = malloc(sizeof(struct _rsc_chn_thread));
    if(pchn == NULL)
    {
	SLOG(ERR,"malloc <_rsc_chn_thread> fail!\n");
	goto create_fail;
    }
    memset(pchn,0,sizeof(*pchn));
    memcpy(&pchn->cattr,attr,sizeof(*attr));
    pchn->chn = chn;
    pchn->cattr.chn = chn;
    pchn->cattr.runflag = 1;
    pchn->cattr.onstart = 0;
    if(pchn->cattr.unit_time_limit<=0)
	pchn->cattr.unit_time_limit = 60*60;
    if(pchn->cattr.unit_size_limit<=0)
	pchn->cattr.unit_size_limit = 800*1024*1024;

    if(pthread_create(&pchn->ctid, NULL, rsc_rec_thread, (void *)pchn) < 0)
    {
	SLOG(ERR,"create record pthread in chn[%d] err!\n",chn);
	goto create_fail;
    }
    return 0;

create_fail:
    if(pchn!=NULL)
    {
	free(pchn);
	pchn = NULL;
    }
    pthread_mutex_lock(&rsc_rec_lock);
    destroy_msg_queue(chn);
    rsc_chn_flag &= ~(1<<chn);
    pthread_mutex_unlock(&rsc_rec_lock);
    return -1;
}

int rsc_chn_start(int chn)
{
    int ret = 0;

    ret = send_msg_cmd(chn,CMD_START_CHN,0,0);
    return ret;
}

int rsc_chn_stop(int chn)
{
    int ret = 0;

    ret = send_msg_cmd(chn,CMD_STOP_CHN,0,0);
    return ret;
}

int rsc_chn_getattr(int chn,struct rsc_channel_attr *attr)
{
    int ret = 0;

    ret = send_msg_cmd(chn,CMD_GETATTR_CHN,0,0);
    if(ret==0)
    {
	int cmd = CMD_RES_GETATTR;
	int size = sizeof(*attr);
	int cnt = 5;

	do{
	    usleep(100000);
	    ret = recv_msg_cmd(chn,&cmd,(char*)attr,&size);
	}while(cnt-- && ret);
    }
    return ret;
}

int rsc_chn_setattr(int chn,struct rsc_channel_attr *attr)
{
    int ret = 0;

    ret = send_msg_cmd(chn,CMD_SETATTR_CHN,(char*)attr,sizeof(*attr));
    return ret;
}

int rsc_chn_getstat(int chn,struct rsc_channel_stat *cstat)
{
    int ret = 0;

    ret = send_msg_cmd(chn,CMD_GETSTAT_CHN,0,0);
    if(ret==0)
    {
	int cmd = CMD_RES_GETSTAT;
	int size = sizeof(*cstat);
	int cnt = 5;

	do{
	    usleep(100000);
	    ret = recv_msg_cmd(chn,&cmd,(char*)cstat,&size);
	}while(cnt-- && ret);
    }
    return ret;
}

int rsc_chn_destroy(int chn)
{
    int ret = 0;

    ret = send_msg_cmd(chn,CMD_DESTROY_CHN,0,0);
    return ret;
}

